package cn.doitedu.rtdw.dash_board;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/10/1
 * @Desc: 学大数据，上多易教育
 *
 **/
public class Job6_各品牌销售额topn个商品 {
    public static void main(String[] args) throws Exception {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(2);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 开启提前触发器，配合 滚动窗口，实现cumulate窗口的效果
         */
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "5000 ms");


        // 建cdc连接器表，映射mysql中的 订单表
        // 1. 建表映射 业务库中的  oms_order表
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        " id            BIGINT ,          " +
                        " status        INT ,             " +
                        " modify_time   timestamp(3)  ,   " +
                        " rt as modify_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED                 " +
                        " ) WITH (                                      " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );

        // 2. 建表映射 业务库中的  oms_order_item 表
        tenv.executeSql(
                "CREATE TABLE order_item_mysql (    " +
                        " id                BIGINT ,         " +
                        " order_id          BIGINT ,         " +
                        " product_id        BIGINT ,         " +
                        " product_name      string  ,        " +
                        " product_brand     string  ,        " +
                        " real_amount       decimal(10,2),   " +
                        " primary key (id) not enforced      " +
                        " ) WITH (                                     " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order_item'          " +
                        ")"
        );


        // 3. 双流join，关联订单主表和订单详情表
        tenv.executeSql(
                " create temporary view joined_view as  "+
                        " select                                "+
                        "     o.id as order_id,                 "+
                        "     o.status as order_status,         "+
                        "     o.modify_time,                    "+
                        "     o.rt,                             "+
                        "     i.product_id,                     "+
                        "     i.product_name,                   "+
                        "     i.product_brand,                  "+
                        "     i.real_amount                     "+
                        " from order_mysql o                    "+
                        " join order_item_mysql i               "+
                        " on o.id=i.order_id  and o.status>0    "

        );

        //tenv.executeSql("select * from joined_view").print();


        /**
         * 双流join后，源表中的rt字段，就失去了 时间属性
         * 不能再用于开 time window
         */
//        tenv.executeSql(
//                "select\n" +
//                "    window_start,\n" +
//                "\twindow_end,\n" +
//                "\tproduct_brand,\n" +
//                "\tproduct_id,\n" +
//                "\tproduct_name,\n" +
//                "\tsum(real_amount) as real_amount\n" +
//                "\t\n" +
//                "from table(\n" +
//                "    cumulate(table joined_view,descriptor(rt),interval '5' second ,interval '24' hour)\n" +
//                ")\n" +
//                "group by \n" +
//                "    window_start,\n" +
//                "\twindow_end,\n" +
//                "\tproduct_brand,\n" +
//                "\tproduct_id,\n" +
//                "\tproduct_name").print();


        /**
         * 曲线救国：  为join后的数据，重新定义  watermark
         */
        DataStream<Row> changelogStream = tenv.toChangelogStream(tenv.from("joined_view"));
        // 本流->view 的方法，只支持  appendOnly流
//        tenv.createTemporaryView("tmp",changelogStream,
//                Schema.newBuilder()
//                        .column("order_id", DataTypes.BIGINT())
//                        .column("order_status", DataTypes.INT())
//                        .column("modify_time", DataTypes.TIMESTAMP(3))
//                        .column("rt", DataTypes.TIMESTAMP(3))
//                        .column("product_id", DataTypes.BIGINT())
//                        .column("product_name", DataTypes.STRING())
//                        .column("product_brand", DataTypes.STRING())
//                        .column("real_amount", DataTypes.DECIMAL(10,2))
//                        .columnByExpression("rt2","modify_time")
//                        .watermark("rt2","rt2 - interval '0' second")
//                .build());

        Table table = tenv.fromChangelogStream(changelogStream,Schema.newBuilder()
                .column("order_id", DataTypes.BIGINT())
                .column("order_status", DataTypes.INT())
                .column("modify_time", DataTypes.TIMESTAMP(3))
                .column("rt", DataTypes.TIMESTAMP(3))
                .column("product_id", DataTypes.BIGINT())
                .column("product_name", DataTypes.STRING())
                .column("product_brand", DataTypes.STRING())
                .column("real_amount", DataTypes.DECIMAL(10,2))
                .columnByExpression("rt2","modify_time")
                .watermark("rt2","rt2 - interval '0' second")
                .build());

        tenv.createTemporaryView("tmp",table);

        /**
         *
         */

        tenv.executeSql(
                " select                                                                                                                "+
                        "     product_brand,                                                                                                    "+
                        "     product_id,                                                                                                       "+
                        "     product_name,                                                                                                     "+
                        "     real_amount,                                                                                                      "+
                        " 	  rn                                                                                                                "+
                        " FROM (                                                                                                                "+
                        "     select                                                                                                            "+
                        "         window_start,                                                                                                 "+
                        "         window_end,                                                                                                   "+
                        "         product_brand,                                                                                                "+
                        "         product_id,                                                                                                   "+
                        "         product_name,                                                                                                 "+
                        "         real_amount,                                                                                                  "+
                        "         row_number() over(partition by window_start,window_end,product_brand order by real_amount desc ) as rn        "+
                        "     FROM (                                                                                                            "+
                        "         select                                                                                                        "+
                        "           TUMBLE_START(rt2,INTERVAL '24' HOUR) as window_start,                                                       "+
                        "           TUMBLE_END(rt2,INTERVAL '24' HOUR) as window_end,                                                           "+
                        "         	product_brand,                                                                                              "+
                        "         	product_id,                                                                                                 "+
                        "         	product_name,                                                                                               "+
                        "         	sum(real_amount) as real_amount                                                                             "+
                        "         from  tmp                                                                                                     "+
                        "         group by                                                                                                      "+
                        "           TUMBLE(rt2,INTERVAL '24' HOUR),                                                                             "+
                        "         	product_brand,                                                                                              "+
                        "         	product_id,                                                                                                 "+
                        "         	product_name                                                                                                "+
                        "     ) o1	                                                                                                            "+
                        " ) o2                                                                                                                  "+
                        " WHERE rn<=10                                                                                                          "
        ).print();





        env.execute();



    }
}
